package mqConsumer

import (
	"encoding/json"
	"github.com/Sir-yuChen/funAnalyse/global"
	"github.com/Sir-yuChen/funAnalyse/model"
	"github.com/Sir-yuChen/funAnalyse/utils/kafka"
	"github.com/Sir-yuChen/funAnalyse/utils/kafka/spec"
	"go.uber.org/zap"
	"time"
)

type BulletinKafkaConsumerHandle struct{}

// Consumer 消费公共消息 所有的消费实现类继承该方法即可
func (b *BulletinKafkaConsumerHandle) Consumer(kmsg spec.Msg) {
	global.GVA_LOG.Info("kafka消费公告", zap.Any("kmsg", string(kmsg.Data())))
	var kaJson kafka.MsgDataRequest
	var msg BulletinMsg
	if string(kmsg.Data()) == "" {
		global.GVA_LOG.Error("BulletinKafkaConsumerHandle 接口到的消息为空", zap.Any("msgKey", kmsg.GetKey()))
		return
	}
	//1.json转map
	err := json.Unmarshal(kmsg.Data(), &kaJson)
	if err == nil {
		err = json.Unmarshal([]byte(kaJson.Content), &msg)
	}
	if err != nil {
		global.GVA_LOG.Error("BulletinKafkaConsumerHandle 数据转换异常", zap.Any("kmsg", kmsg.Data()), zap.Error(err))
		return
	}
	var total int64
	//校验消息是否重复 同一条消息对同一个人在当天只能推送一次 select * from stock_user_bulletin where date(create_time) = curdate();
	err = global.GVA_DB.Model(model.StockUserBulletin{}).Where(" customer_id = ? AND bulletin_id = ? AND date(create_time) = curdate()", msg.UserId, msg.BulletinId).Count(&total).Error
	if err != nil {
		global.GVA_LOG.Error("BulletinKafkaConsumerHandle 校验当天推送消息是否重复", zap.Any("BulletinMsg", msg), zap.Error(err))
		return
	}
	if total > 0 {
		global.GVA_LOG.Warn("BulletinKafkaConsumerHandle 用户当天已推送过该消息,注意无需再次推送", zap.Any("BulletinMsg", msg))
		return
	}
	now := model.LocalTime(time.Now())
	userBulletin := model.StockUserBulletin{
		CustomerId:  msg.UserId,
		BulletinId:  int64(msg.BulletinId),
		Status:      1,
		CreateTime:  &now,
		LastUpdTime: &now,
	}
	//新增并推送
	if err := global.GVA_DB.Model(&userBulletin).Create(&userBulletin).Error; err != nil {
		global.GVA_LOG.Error("BulletinKafkaConsumerHandle 新增用户与消息之间关系异常", zap.Any("BulletinMsg", msg), zap.Error(err))
		return
	}
}

func (b *BulletinKafkaConsumerHandle) FailAction(err *kafka.ConsumeError) {
	global.GVA_LOG.Error("BulletinKafkaConsumerHandle  消费消息异常 执行 FailAction", zap.Any("err", err.Err))
}
